package com.bigdata.warehouse.dwd;

import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdCarsLog {
    public static void main(String[] args) {
        //1.获取Stream的执行环境
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);

        //2.创建表的执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);

        //开启checkpoint容错
//        senv.enableCheckpointing(60000);
//        senv.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        senv.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
//        senv.getCheckpointConfig().setCheckpointTimeout(10000);
//        senv.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        //设置状态后端
        //（1）开启RocksDB
//        senv.setStateBackend(new EmbeddedRocksDBStateBackend());
//        //（2）设置checkpoint存储位置
//        senv.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://mycluster/flink/checkpoints"));

        //3.读取车辆进出事实表
        tEnv.executeSql("CREATE TABLE ods_cars_log (\n" +
                " id STRING,\n" +
                " opTime STRING,\n" +
                " ctype SMALLINT,\n" +
                " carCode STRING,\n" +
                " cId BIGINT,\n" +
                " proc_time as PROCTIME()\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'ods_cars_log',\n" +
                " 'properties.bootstrap.servers' = 'master:9092',\n" +
                " 'properties.group.id' = 'ods_cars_log',\n" +
                " 'scan.startup.mode' = 'earliest-offset',\n" +
                " 'format' = 'json'\n" +
                ")\n");

        //4.读取车辆维度表
        tEnv.executeSql("CREATE TABLE dim_base_cars(\n" +
                " id INT,\n" +
                " owerId INT,\n" +
                " carCode STRING,\n" +
                " carColor STRING,\n" +
                " type TINYINT,\n" +
                " remark STRING,\n" +
                " PRIMARY KEY(id) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive',\n" +
                " 'table-name' = 'dim_base_cars'\n" +
                ")\n");

        //5.事实表与维度表进行关联
        Table resultTable = tEnv.sqlQuery("select "+
                "cl.id, "+
                "c.owerId, "+
                "cl.opTime, "+
                "cl.cId, "+
                "cl.carCode, "+
                "cl.ctype "+
                "from ods_cars_log cl "+
                "left join dim_base_cars for system_time as of cl.proc_time as c "+
                "on cl.carCode=c.carCode"
        );
        tEnv.createTemporaryView("resultTable",resultTable);

        //6.创建dwd_cars_log表
        tEnv.executeSql("CREATE TABLE dwd_cars_log (\n" +
                " id STRING,\n" +
                " owerId INT,\n" +
                " opTime STRING,\n" +
                " cId BIGINT,\n" +
                " carCode STRING,\n" +
                " ctype SMALLINT,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'upsert-kafka',\n" +
                " 'topic' = 'dwd_cars_log',\n" +
                " 'properties.bootstrap.servers' = 'master:9092',\n" +
                " 'key.format' = 'json',\n" +
                " 'value.format' = 'json'\n" +
                ")\n");

        //7.将flink 查询结果写入dwd_cars_log
        tEnv.executeSql("insert into dwd_cars_log select * from resultTable");
    }
}
